package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.TradeSubjectChapterCourseProvinceSourceOrderBean;
import com.atguigu.edu.realtime.util.ClickHouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class DwsTradeSubjectChapterCourseProvinceSourceOrderWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);


        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_subject_chapter_course_province_source_order_window";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> orderSource = env.addSource(kafkaConsumer);

        SingleOutputStreamOperator<String> filterDS = orderSource.filter(
                new FilterFunction<String>() {
                    @Override
                    public boolean filter(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);

                        String userId = jsonObj.getString("user_id");


                        return userId != null;


                    }
                }
        );

        SingleOutputStreamOperator<JSONObject> jsonObjStream = filterDS.map(JSON::parseObject);

        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(r -> r.getString("id"));
        //根据orderId来排序

        SingleOutputStreamOperator<JSONObject> processedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> lastValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        lastValueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<JSONObject>("last_value_state", JSONObject.class)
                        );
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject lastValue = lastValueState.value();
                        if (lastValue == null) {
                            long currentProcessingTime = ctx.timerService().currentProcessingTime();
                            ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                            lastValueState.update(jsonObj);
                        } else {
                            String lastRowOpTs = lastValue.getString("row_op_ts");
                            String rowOpTs = jsonObj.getString("row_op_ts");
                            if (TimestampLtz3CompareUtil.compare(lastRowOpTs, rowOpTs) <= 0) {
                                lastValueState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws IOException {
                        JSONObject lastValue = this.lastValueState.value();
                        if (lastValue != null) {
                            out.collect(lastValue);
                        }
                        lastValueState.clear();
                    }
                }
        );



        // TODO 7. 转换数据结构
        SingleOutputStreamOperator<TradeSubjectChapterCourseProvinceSourceOrderBean> javaBeanStream = processedStream.map(
                jsonObj -> {
                    String orderId = jsonObj.getString("order_id");
                    String userId = jsonObj.getString("user_id");
                    String courseId = jsonObj.getString("course_id");
                    String provinceId = jsonObj.getString("province_id");
                    Double originalAmount = jsonObj.getDouble("original_amount");
                    Double couponReduce = jsonObj.getDouble("coupon_reduce");
                    Double finalAmount = jsonObj.getDouble("final_amount");
                    String sourceId = jsonObj.getString("source_id");
                    Long ts = jsonObj.getLong("ts") * 1000L;
                    TradeSubjectChapterCourseProvinceSourceOrderBean trademarkUserOrderBean = TradeSubjectChapterCourseProvinceSourceOrderBean.builder()
                            .orderIdSet(new HashSet<String>(
                                    Collections.singleton(orderId)
                            ))
                            .courseId(courseId)
                            .userId(userId)
                            .orderUuCount(0L)
                            .provinceId(provinceId)
                            .originalAmount(originalAmount)
                            .couponReduce(couponReduce == null ? 0.0 : couponReduce)
                            .finalAmount(finalAmount == null ? 0.0 : finalAmount)
                            .sourceId(sourceId)
                            .ts(ts)
                            .build();
                    return trademarkUserOrderBean;
                }
        );


        // TODO 8. 按照 user_id 分组
        KeyedStream<TradeSubjectChapterCourseProvinceSourceOrderBean, String> userIdKeyedStream = javaBeanStream.keyBy(TradeSubjectChapterCourseProvinceSourceOrderBean::getUserId);


        // TODO 9. 统计下单独立用户数
        SingleOutputStreamOperator<TradeSubjectChapterCourseProvinceSourceOrderBean> withUuCtJavaBean = userIdKeyedStream.process(
                new KeyedProcessFunction<String, TradeSubjectChapterCourseProvinceSourceOrderBean, TradeSubjectChapterCourseProvinceSourceOrderBean>() {

                    private ValueState<String> lastOrderDt;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        ValueStateDescriptor<String> valueStateDescriptor =
                                new ValueStateDescriptor<>("course_window_last_order_dt", String.class);
                        // 设置状态 ttl
                        valueStateDescriptor.enableTimeToLive(
                                StateTtlConfig
                                        .newBuilder(Time.days(1L))
                                        .updateTtlOnCreateAndWrite()
                                        .build()
                        );
                        lastOrderDt = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(TradeSubjectChapterCourseProvinceSourceOrderBean tradeSubjectChapterCourseProvinceSourceOrderBean, Context context, Collector<TradeSubjectChapterCourseProvinceSourceOrderBean> out) throws Exception {
                        Long ts = tradeSubjectChapterCourseProvinceSourceOrderBean.getTs();
                        String curDate = DateFormatUtil.toDate(ts);
                        String lastDt = lastOrderDt.value();
                        if (lastDt == null || !lastDt.equals(curDate)) {
                            tradeSubjectChapterCourseProvinceSourceOrderBean.setOrderUuCount(1L);
                            lastOrderDt.update(curDate);
                        }
                        out.collect(tradeSubjectChapterCourseProvinceSourceOrderBean);
                    }
                }
        );


        // TODO 10. 设置水位线
        SingleOutputStreamOperator<TradeSubjectChapterCourseProvinceSourceOrderBean> withWatermarkDS = withUuCtJavaBean.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeSubjectChapterCourseProvinceSourceOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSubjectChapterCourseProvinceSourceOrderBean>() {

                                    @Override
                                    public long extractTimestamp(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean, long recordTimestamp) {
                                        return javaBean.getTs();
                                    }
                                }
                        )
        );


        // TODO 11. 分组
        KeyedStream<TradeSubjectChapterCourseProvinceSourceOrderBean, String> keyedForAggregateStream = withWatermarkDS.keyBy(
                new KeySelector<TradeSubjectChapterCourseProvinceSourceOrderBean, String>() {

                    @Override
                    public String getKey(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean) throws Exception {
                        return javaBean.getCourseId();
                    }
                }
        );

        // TODO 12. 开窗
        WindowedStream<TradeSubjectChapterCourseProvinceSourceOrderBean, String, TimeWindow> windowDS = keyedForAggregateStream.window(TumblingEventTimeWindows.of(
                org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));





        // TODO 13. 聚合
        SingleOutputStreamOperator<TradeSubjectChapterCourseProvinceSourceOrderBean> reducedStream = windowDS
                .reduce(
                        new ReduceFunction<TradeSubjectChapterCourseProvinceSourceOrderBean>() {
                            @Override
                            public TradeSubjectChapterCourseProvinceSourceOrderBean reduce(TradeSubjectChapterCourseProvinceSourceOrderBean value1, TradeSubjectChapterCourseProvinceSourceOrderBean value2) throws Exception {
                                value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                                value1.setOrderUuCount(value1.getOrderUuCount() + value2.getOrderUuCount());
                                value1.setOriginalAmount(value1.getOriginalAmount() + value2.getOriginalAmount());
                                value1.setCouponReduce(value1.getCouponReduce() + value2.getCouponReduce());
                                value1.setFinalAmount(value1.getFinalAmount() + value2.getFinalAmount());
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<TradeSubjectChapterCourseProvinceSourceOrderBean, TradeSubjectChapterCourseProvinceSourceOrderBean, String, TimeWindow>() {

                            @Override
                            public void process(String key, Context context, Iterable<TradeSubjectChapterCourseProvinceSourceOrderBean> elements, Collector<TradeSubjectChapterCourseProvinceSourceOrderBean> out) throws Exception {

                                String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                                String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                                for (TradeSubjectChapterCourseProvinceSourceOrderBean element : elements) {
                                    element.setStt(stt);
                                    element.setEdt(edt);
                                    element.setOrderCount((long) (element.getOrderIdSet().size()));
                                    element.setTs(System.currentTimeMillis());
                                    out.collect(element);
                                }
                            }
                        }
                );


        SingleOutputStreamOperator<TradeSubjectChapterCourseProvinceSourceOrderBean> withCourseInfoStream = AsyncDataStream.unorderedWait(
                reducedStream,
                new DimAsyncFunction<TradeSubjectChapterCourseProvinceSourceOrderBean>("dim_course_info".toUpperCase()) {

                    @Override
                    public void join(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean, JSONObject jsonObj) throws Exception {
                        javaBean.setCourseName(jsonObj.getString("course_name".toUpperCase()));
                        javaBean.setSubjectId(jsonObj.getString("subject_id".toUpperCase()));

                    }

                    @Override
                    public String getKey(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean) {
                        return javaBean.getCourseId();
                    }
                },
                60 * 5, TimeUnit.SECONDS
        );
        // 14.2 关联 spu_info 表
        SingleOutputStreamOperator<TradeSubjectChapterCourseProvinceSourceOrderBean> withProvinceInfoStream = AsyncDataStream.unorderedWait(
                withCourseInfoStream,
                new DimAsyncFunction<TradeSubjectChapterCourseProvinceSourceOrderBean>("dim_base_province".toUpperCase()) {
                    @Override
                    public void join(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean, JSONObject dimJsonObj) throws Exception {
                        javaBean.setProvinceName(
                                dimJsonObj.getString("name".toUpperCase())
                        );
                    }

                    @Override
                    public String getKey(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean) {
                        return javaBean.getProvinceId();
                    }
                },
                60 * 5, TimeUnit.SECONDS
        );

        // 14.3 关联品牌表 base_trademark
        SingleOutputStreamOperator<TradeSubjectChapterCourseProvinceSourceOrderBean> withSubjectStream = AsyncDataStream.unorderedWait(
                withProvinceInfoStream,
                new DimAsyncFunction<TradeSubjectChapterCourseProvinceSourceOrderBean>("dim_base_subject_info".toUpperCase()) {
                    @Override
                    public void join(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean, JSONObject jsonObj) throws Exception {
                        javaBean.setSubjectName(jsonObj.getString("subject_name".toUpperCase()));
                    }

                    @Override
                    public String getKey(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean) {
                        return javaBean.getSubjectId();
                    }
                },
                5 * 60, TimeUnit.SECONDS
        );




        SingleOutputStreamOperator<TradeSubjectChapterCourseProvinceSourceOrderBean> withSourceStream = AsyncDataStream.unorderedWait(
                withSubjectStream,
                new DimAsyncFunction<TradeSubjectChapterCourseProvinceSourceOrderBean>("dim_base_source".toUpperCase()) {
                    @Override
                    public void join(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean, JSONObject jsonObj) throws Exception {
                        javaBean.setSourceName(jsonObj.getString("source_site".toUpperCase()));
                    }

                    @Override
                    public String getKey(TradeSubjectChapterCourseProvinceSourceOrderBean javaBean) {
                        return javaBean.getSourceId();
                    }
                },
                5 * 60, TimeUnit.SECONDS
        );

        withSourceStream.print("test");

        // TODO 15. 写出到 OLAP 数据库
        SinkFunction<TradeSubjectChapterCourseProvinceSourceOrderBean> jdbcSink =
                ClickHouseUtil.<TradeSubjectChapterCourseProvinceSourceOrderBean>getJdbcSink(
                        "insert into dws_trade_subject_chapter_course_province_source_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
                );
        withSourceStream.<TradeSubjectChapterCourseProvinceSourceOrderBean>addSink(jdbcSink);

        env.execute();




}

}
